using System;
using System.Collections.Generic;
using Me.Prettyprint.Cassandra.Utils;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Ddl;
using Me.Prettyprint.Hector.Api.Exceptions;
using Org.Apache.Cassandra.Thrift;

namespace Me.Prettyprint.Cassandra.Service
{
    public class ThriftCluster : AbstractCluster, Cluster
    {
        public ThriftCluster(string clusterName, CassandraHostConfigurator cassandraHostConfigurator)
            : base(clusterName, cassandraHostConfigurator)
        {
        }

        public ThriftCluster(string clusterName, CassandraHostConfigurator cassandraHostConfigurator, Dictionary<string, string> credentials)
            : base(clusterName,cassandraHostConfigurator)
        {
        }
        public override void OnStartup()
        {
            if (Configurator.RunAutoDiscoveryAtStartup)
            {
                ConnectionManager.DoAddNodes();
            }
        }
        private ThriftCfDef ToThriftImplementation(ColumnFamilyDefinition cfdef) 
        {
            if(cfdef is ThriftCfDef)
            {
                return (ThriftCfDef) cfdef;
            }
            else
            {
                return new ThriftCfDef(cfdef);
            }
        }
        private ThriftKsDef ToThriftImplementation(KeyspaceDefinition cfdef)
        {
            if(cfdef is ThriftKsDef) 
            {
                return (ThriftKsDef) cfdef;
            }
            else
            {
                return new ThriftKsDef(cfdef);
            }
        }
        public new List<TokenRange> DescribeRing(string keyspace)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, List<TokenRange>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    List<TokenRange> result = cassandra.describe_ring(keyspace);
                    return result;
                }
                catch (Exception e)
                {
                    logger.Error(e.Message);
                    throw exceptionTranslator.translate(e);
                }
            };
            Operation<List<TokenRange>> op = new OperationBuilder<List<TokenRange>>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            ConnectionManager.operateWithFailover(op);
            return op.Result;
        }
        public override Dictionary<string,List<String>> DescribeSchemaVersions()
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, Dictionary<string,List<String>>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    Dictionary<string,List<String>> result = cassandra.describe_schema_versions();
                    return result;
                }
                catch (Exception e)
                {
                    logger.Error(e.Message);
                    throw exceptionTranslator.translate(e);
                }
            };
            Operation<Dictionary<string,List<String>>> op = new OperationBuilder<Dictionary<string,List<String>>>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            ConnectionManager.operateWithFailover(op);
            return op.Result;
        }
        public override string UpdateKeyspace(KeyspaceDefinition ksdef)
        {
            return UpdateKeyspace(ksdef, false);
        }
        public override string UpdateKeyspace(KeyspaceDefinition ksdef, bool waitForSchemaAgreement)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, string> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    if (waitForSchemaAgreement)
                    {
                        WaitForSchemaAgreement(cassandra);
                    }
                    string schemaId = cassandra.system_update_keyspace(ToThriftImplementation(ksdef).ToThrift());
                    if (waitForSchemaAgreement)
                    {
                        WaitForSchemaAgreement(cassandra);
                    }
                    return schemaId;
                }
                catch (Exception e)
                {
                    logger.Error(e.Message);
                    throw exceptionTranslator.translate(e);
                }
            };
            Operation<string> op = new OperationBuilder<string>(x);
            op.OperationType = OperationType.META_WRITE;
            op.FailoverPolicy = FailoverPolicy.FAIL_FAST;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            return op.Result;
        }
        public override string AddColumnFamily(ColumnFamilyDefinition cfdef)
        {
            return AddColumnFamily(cfdef, false);
        }
        public override string AddColumnFamily(ColumnFamilyDefinition cfdef, bool waitForSchemaAgreement) 
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, string> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    if (waitForSchemaAgreement)
                    {
                        WaitForSchemaAgreement(cassandra);
                    }
                    string schemaId = cassandra.system_add_column_family(ToThriftImplementation(cfdef).ToThrift());
                    if (waitForSchemaAgreement)
                    {
                        WaitForSchemaAgreement(cassandra);
                    }
                    return schemaId;
                }
                catch (Exception e)
                {
                    logger.Error(e.Message);
                    throw exceptionTranslator.translate(e);
                }
            };
            Operation<string> op = new OperationBuilder<string>(x);
            op.OperationType = OperationType.META_WRITE;
            op.FailoverPolicy = FailoverPolicy.FAIL_FAST;
            op.KeyspaceName = cfdef.KeyspaceName;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            return op.Result;
        }
        public override string UpdateColumnFamily(ColumnFamilyDefinition cfdef)
        {
            return UpdateColumnFamily(cfdef, false);
        }
        public override string UpdateColumnFamily(ColumnFamilyDefinition cfdef, bool waitForSchemaAgreement)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, string> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    if (waitForSchemaAgreement)
                    {
                        WaitForSchemaAgreement(cassandra);
                    }
                    string schemaId = cassandra.system_update_column_family(ToThriftImplementation(cfdef).ToThrift());
                    if (waitForSchemaAgreement)
                    {
                        WaitForSchemaAgreement(cassandra);
                    }
                    return schemaId;
                }
                catch (Exception e)
                {
                    logger.Error(e.Message);
                    throw exceptionTranslator.translate(e);
                }
            };
            Operation<string> op = new OperationBuilder<string>(x);
            op.OperationType = OperationType.META_WRITE;
            op.FailoverPolicy = FailoverPolicy.FAIL_FAST;
            op.KeyspaceName = cfdef.KeyspaceName;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            return op.Result;
        }
        public override string AddKeyspace(KeyspaceDefinition ksdef)
        {
            return AddKeyspace(ksdef, false);
        }
        public override string AddKeyspace(KeyspaceDefinition ksdef, bool waitForSchemaAgreement)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, string> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    if (waitForSchemaAgreement)
                    {
                        WaitForSchemaAgreement(cassandra);
                    }
                    string schemaId = cassandra.system_add_keyspace(ToThriftImplementation(ksdef).ToThrift());
                    if (waitForSchemaAgreement)
                    {
                        WaitForSchemaAgreement(cassandra);
                    }
                    return schemaId;
                }
                catch (Exception e)
                {
                    logger.Error(e.Message);
                    throw exceptionTranslator.translate(e);
                }
            };
            Operation<string> op = new OperationBuilder<string>(x);
            op.OperationType = OperationType.META_WRITE;
            op.FailoverPolicy = FailoverPolicy.FAIL_FAST;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            return op.Result;
        }
    }
}